package com.exm.collectcodenew.kafka.consumer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.util.Arrays;
import java.util.Properties;
public class ConsumerTest {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "59.110.35.177:9092");//必须指定
        props.put("group.id","test-group");//必须指定
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");//必须指定
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");//必须指定
        props.put("enable.auto.commit","true");
        props.put("auto.commit.interval.ms","1000");
        props.put("auto.offset.reset","earliest");//从最早的消息开始读取
        // 添加拦截器配置
        props.put("interceptor.classes", "com.exm.collectcodenew.kafka.producer.customInterceptor.LoggingOffsetInterceptor");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        //创建consumer实例
        consumer.subscribe(Arrays.asList("topic-test1"));
        //注意：subscribe方法不是增量式的，这意味着后续的subscribe调用会完全覆盖之前的订阅语句。
        consumer.subscribe(Arrays.asList("topic-test2"));
        while(true){
            //通常情况下如果consumer 拿到了足够多的可用数据，那么它可以立即从该方法返回:但若当前没有足够多的数据可供返回，consumer会处于阻塞状态。这个超时参数即控制阻塞的最大时间。这里的1000表示即使没有那么多数据，consumer最多也不要等待超过1秒的时间。
            ConsumerRecords<String,String> records=consumer.poll(1000);
            for (ConsumerRecord<String, String> record: records){
                System.out.printf("offset=%d,key=%s,value=%s%n",record.offset(),record.key(),record.value());
            }
        }
    }
}
